package com.yjjxt

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Hello29Broadcast {
  def main(args: Array[String]): Unit = {
    val sparkContext = new SparkContext(new SparkConf().setMaster("local").setAppName("Hello29Broadcast"))

    /**
     * 这种实现方式，一共有4个Task，每个Task都要拥有List变量
     * Task属于线程级别，线程可以共享进程的数据
     * 广播的数据下游是不会修改的
     * 我们直接将广播的数据存放到进程 可以所有的Task使用
     */
    val lines1 = sparkContext.parallelize(Array("吃", "喝", "玩", "跳"), 4)
    val list1 = List("吃", "喝", "睡")
    val filter1: RDD[String] = lines1.filter(ele => {
      list1.contains(ele)
    })
    filter1.foreach(println)

    /**
     * 开始使用广播变量
     * 广播变量被存放到Executor的BlockManager空间中,一个Executor只有一个BlockManager
     */
    val lines2 = sparkContext.parallelize(Array("吃2", "喝2", "玩2", "跳2"), 4)
    val list21 = List("吃2", "喝2", "睡2")
    val broadcast: Broadcast[List[String]] = sparkContext.broadcast(list21)
    val filter2: RDD[String] = lines2.filter(ele => {
      broadcast.value.contains(ele)
    })
    filter2.foreach(println)

  }
}
